import pandas as pd
import pymysql
from minio import Minio
from hdfs import InsecureClient
import cryptography
import io
import re


def col_mysql(host, port, username, password, database, table='', getdata=False):
    # 返回数据
    if getdata:
        try:
            connection = pymysql.connect(host=host, port=port, user=username, password=password, database=database)
            # SQL查询语句
            query = 'SELECT * FROM ' + table
            # 执行查询并将结果存储为DataFrame
            df = pd.read_sql_query(query, connection)
            connection.close()
            return df
        except pymysql.Error as e:
            print('Mysql读取异常',str(e))
            return None
    
    # 查询表
    try:
        connection = pymysql.connect(host=host, port=port, user=username, password=password, database=database)
        # 创建游标对象
        cursor = connection.cursor()
        # 执行SQL查询获取表列表
        cursor.execute("SHOW TABLES")
        # 获取查询结果
        tables = cursor.fetchall()
        connection.close()
        result = []
        for item in tables:
            result.append(item[0])
        return result
    except pymysql.Error as e:
        print('conn.py line44 mysql查询异常',str(e))
        return None



def get_hdfs_tree_paths(path, url, user, paths=[]):
    client = InsecureClient(url, user)
    items = client.list(path)
    for item in items:
        item_path = path + item
        if client.status(item_path)['type'] == 'DIRECTORY':
            get_hdfs_tree_paths(item_path + '/', url, user, paths)
        else:
            paths.append(item_path)
    return paths


def col_hdfs(hdfs_host, hdfs_port, user, test=False, hdfs_path='/'):
    url = f'http://{hdfs_host}:{hdfs_port}'
    # 连接客户端
    if test:
        try:
            client = InsecureClient(url, user)
            if client.status('/') is not None:
                paths = []
                get_hdfs_tree_paths('/', url, user,paths)
                return paths
            else:
                return None
        except Exception as err:
            print("HDFS连接异常:", str(err))
            return None
    # 读取文件
    else:
        try:
            client = InsecureClient(url, user=user)
            with client.read(hdfs_path) as f:
                file_data = f.read()
                file_obj = io.BytesIO(file_data)
                encodings_to_try = [
                    'gbk','utf-8','gb18030','gb2312','ascii', 'big5'
                ]
                for encoding_to_try in encodings_to_try:
                    try:
                        file_obj.seek(0)
                        if hdfs_path.endswith('txt'):
                            # 处理txt文件
                            df = pd.read_csv(file_obj,delimiter='\t', encoding=encoding_to_try)
                            df_new = df[df.columns[0]].str.split(r"',|\",", expand=True)
                            #把列名里的\t和"和'去掉
                            df.rename(columns=lambda x: re.sub(r'[\"\t\']', '',x), inplace=True)
                            #把列名按照,切割成多列
                            df_new.columns =df.columns[0].split(',')
                            #去掉所有的'和"
                            df_new=df_new.replace(['\'', '\"'], '', regex=True)
                            return df_new
                            # return df
                        elif hdfs_path.endswith('csv'):
                            # 处理csv文件
                            df = pd.read_csv(file_obj, encoding=encoding_to_try)
                            df=df.replace(['\'', '\"'], '', regex=True)
                            return df
                        elif hdfs_path.endswith('xls') or hdfs_path.endswith('xlsx'):
                            # 处理xls和xlsx文件
                            df = pd.read_excel(file_obj)
                            return df
                    except UnicodeDecodeError:
                        print(encoding_to_try)
                        continue
                return None

        except Exception as err:
            print("HDFS异常:", str(err))
            return None


# 递归操作
def list_files_dfs_minio(client, bucket_name, object_name, files_list):
    try:
        objects = client.list_objects(bucket_name, prefix=object_name, recursive=True)
        for obj in objects:
            if obj.is_dir:
                list_files_dfs_minio(client, bucket_name, obj.object_name + '/', files_list)
            else:
                files_list.append(obj.object_name)
    except Exception as err:
        print("遍历文件失败:", str(err))
        return None


def col_minio(minio_host, minio_port, access_key, secret_key, bucket, test=False, object_name='/', use_ssl=False):
    if test:
        try:
            # 创建 MinIO 客户端对象
            minio_endpoint = minio_host + ':' + minio_port
            minio_client = Minio(minio_endpoint, access_key=access_key,
                                 secret_key=secret_key, secure=use_ssl)
            try:
                files_list = []
                list_files_dfs_minio(minio_client, bucket, object_name, files_list)
                if files_list == []:
                    return None
                else:
                    return files_list
            except Exception as err:
                print("Minio遍历文件失败:", str(err))
                return None
        except Exception as err:
            print("col_minio Minio数据源异常:", str(err))
            return None
    else:
        try:
            # 创建 MinIO 客户端对象
            minio_endpoint = minio_host + ':' + minio_port
            minio_client = Minio(minio_endpoint, access_key=access_key,
                                 secret_key=secret_key, secure=use_ssl)
        except Exception as err:
            print("Minio数据源异常:", str(err))
            return None
        try:
            # 获取文件的数据
            data = minio_client.get_object(bucket, object_name)
            file_data = data.read()  # data.read()是minio的接口，读取出cvs文件的二进制流file_data
            # 把二进制流file_data 转换成文件
            file_obj = io.BytesIO(file_data)
            # 读取文本文件内容
            encodings_to_try = [
                'gbk','utf-8','gb18030','gb2312','ascii', 'big5'
            ]
            for encoding_to_try in encodings_to_try:
                try:
                    file_obj.seek(0)
                    if object_name.endswith('txt'):
                        # 处理txt文件
                        # df = pd.read_csv(file_obj,sep=",", quotechar="'", encoding=encoding_to_try)
                        df = pd.read_csv(file_obj,delimiter='\t', encoding=encoding_to_try)
                        df_new = df[df.columns[0]].str.split(r"',|\",", expand=True)
                        #把列名里的\t和"和'去掉
                        df.rename(columns=lambda x: re.sub(r'[\"\t\']', '',x), inplace=True)
                        #把列名按照,切割成多列
                        df_new.columns =df.columns[0].split(',')
                        #去掉所有的'和"
                        df_new=df_new.replace(['\'', '\"'], '', regex=True)
                        return df_new
                        # return df
                    elif object_name.endswith('csv'):
                        # 处理csv文件
                        df = pd.read_csv(file_obj, encoding=encoding_to_try)
                        df=df.replace(['\'', '\"'], '', regex=True)
                        return df
                    elif object_name.endswith('xls') or object_name.endswith('xlsx'):
                        # 处理xls和xlsx文件
                        df = pd.read_excel(file_obj)
                        return df
                except UnicodeDecodeError:
                    print(encoding_to_try)
                    continue
            return None
        except Exception as err:
            print("Minio读取文件异常:", str(err))
            return None

# 以下是数据文件采集接口
def col_file(file, sep=",", quotechar="'"):
    print(file)
    # 按优先级列出要尝试的编码
    encodings_to_try = [
        'utf-8', 'gb18030', 'gbk', 'big5', 'shift_jis', 'euc-jp', 'euc-kr', 'iso-8859-1', 'windows-1252'
    ]
    # 检查文件格式
    allowed_formats = ['csv', 'txt', 'xlsx']
    file_type = file.name.split('.')[-1]
    if file_type not in allowed_formats:
        return None
    for encoding_to_try in encodings_to_try:
        try:
            if file_type == "txt":
                dataframe = pd.read_csv(file, sep=sep, quotechar=quotechar, encoding=encoding_to_try)
                return dataframe
            elif file_type == "csv":
                dataframe = pd.read_csv(file, encoding=encoding_to_try)
                return dataframe
            elif file_type == 'xlsx':
                dataframe = pd.read_excel(file)
                return dataframe

        except UnicodeDecodeError:
            continue
    return None
